package com.pxene.caw;

import java.io.IOException;
import java.net.URL;
import java.net.URLDecoder;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

public class ClearDataMapper extends Mapper<LongWritable, Text, LongWritable, Text> {
	private MultipleOutputs<LongWritable, Text> mos ;
	private Map<String, String> map=new HashMap<>();
	
	private String out="/user/vendorwins/chenjinghui/";

	private URL urlParser;
	
	
	@Override
	protected void setup(Mapper<LongWritable, Text, LongWritable, Text>.Context context) throws IOException, InterruptedException {
//		map.putAll(Constant.carMap);
//		map.putAll(Constant.trainMap);
//		map.putAll(Constant.flightMap);
		map.putAll(Constant.posMap);
		mos= new MultipleOutputs<LongWritable, Text>(context);
	}
	
	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		String line=value.toString();
		String[] words = line.split("\\|");
		if(words.length!=38){
			return ;
		}
		
		String destinationURL = null;
		
		destinationURL = URLDecoder.decode(words[27].trim().replaceAll("%(?![0-9a-fA-F]{2})", "%25"), "UTF-8");
		Set<String> keys = map.keySet();
		for (String each : keys) {
			if(destinationURL.matches(map.get(each))){
				mos.write(key, new Text(destinationURL),out+each );
			}
		}
		
	}
	

}
